Creating Custom Agents Tutorial

Overview

This tutorial walks through creating a simple diagnostic agent that monitors sensor data and detects when values exceed thresholds. The agent demonstrates the three thread architecture without complex ML models.

What you'll build: A threshold monitoring agent that:

Ingests sensor data from Kafka
Stores data to MySQL database
Checks values against configured thresholds
Publishes alerts when thresholds are exceeded

Prerequisites

SMOCS repository cloned
Docker and Docker Compose installed
Basic Python knowledge
Understanding of SMOCS architecture (see Architecture section)

Step 1: Plan Your Agent

Agent Responsibilities

Data Ingest Thread:

Subscribe to sensor data topic
Parse numeric sensor values
Store to database with timestamps

ML Training Thread:

Calculate statistics from historical data
Update dynamic thresholds
Publish threshold updates

ML Inference Thread:

Check incoming values against thresholds
Publish alerts for violations
Track violation frequency

Configuration Design

threshold_agent1:
  # Threading
  enabled_threads: ['ingest', 'training', 'inference']
  
  # Data specification
  model_input:
    channels:
      - temperature
      - pressure
      - flow_rate
    bounds:
      - [0.0, 100.0]
      - [0.0, 50.0]
      - [0.0, 10.0]
  
  # Thresholds
  static_thresholds:
    temperature: 80.0
    pressure: 45.0
    flow_rate: 9.0
  
  use_dynamic_thresholds: true
  threshold_percentile: 95.0
  
  # Topics
  kafka_topics:
    input: "sensor-data"
    output: "threshold-alerts"
    training_output: "threshold-updates"

Step 2: Create Agent File

Create smocs/agents/threshold_agent.py:

import os
import json
import time
import logging
import argparse
import numpy as np
from typing import Dict, Any, Optional, List, Tuple
from datetime import datetime

from smocs.cores import AgentBase, DataIngestThreadBase, MLTrainingThreadBase, MLInferenceThreadBase
from smocs.utils import ConfigLoader, ChannelFilter, setup_logging

Step 3: Implement Data Ingest Thread

class ThresholdDataIngestThread(DataIngestThreadBase):
    """
    Ingests sensor data and stores to database.
    """
    
    def store_message(self, message_data, topic, partition, offset) -> bool:
        """
        Parse sensor message and store to database.
        
        Args:
            message_data: Filtered message dict with 'channels' and 'timestamp'
            topic: Kafka topic name
            partition: Kafka partition
            offset: Kafka offset
            
        Returns:
            True if stored successfully, False otherwise
        """
        try:
            # Extract timestamp
            if 'timestamp' in message_data:
                timestamp = datetime.fromtimestamp(message_data['timestamp'])
            else:
                timestamp = datetime.now()
            
            # Get filtered channels (already processed by base class)
            channels = message_data.get('channels', {})
            
            if not channels:
                logging.error("No channels in message")
                return False
            
            # Convert to numpy array
            channel_values = np.array(list(channels.values()), dtype=np.float32)
            
            # Store to database
            sensor_data = {
                'state_source_timestamp': timestamp.strftime('%Y-%m-%d %H:%M:%S.%f'),
                'state_received_timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'),
                'state': channel_values
            }
            
            status = self.db_manager.record_sensor_data(sensor_data)
            
            if status == 0:
                logging.debug(f"Stored sensor data: {len(channel_values)} channels")
                return True
            else:
                logging.error(f"Failed to store sensor data, status: {status}")
                return False
                
        except Exception as e:
            logging.error(f"Error storing message: {e}")
            return False

Key points:

message_data already contains filtered channels from base class
Convert channel values to numpy array
Use record_sensor_data() with proper timestamp format
Return True for success, False for failure

Step 4: Implement Training Thread

class ThresholdMLTrainingThread(MLTrainingThreadBase):
    """
    Calculates dynamic thresholds from historical data.
    """
    
    def __init__(self, agent_id: str, config: Dict[str, Any]):
        self.use_dynamic = config.get('use_dynamic_thresholds', False)
        self.percentile = config.get('threshold_percentile', 95.0)
        self.static_thresholds = config.get('static_thresholds', {})
        self.min_samples = config.get('min_training_samples', 1000)
        
        self.current_thresholds = self.static_thresholds.copy()
        self.last_training_count = 0
        
        super().__init__(agent_id, config)
    
    def build_model(self):
        """Not needed for threshold agent."""
        pass
    
    def get_training_data(self) -> Optional[np.ndarray]:
        """
        Retrieve recent sensor data from database.
        
        Returns:
            Array of sensor readings or None if insufficient data
        """
        try:
            total_samples = self.db_manager.get_size("agent_inferences")
            
            logging.info(f"Database contains {total_samples} samples (need {self.min_samples})")
            
            if total_samples < self.min_samples:
                return None
            
            if total_samples <= self.last_training_count:
                logging.debug("No new data since last training")
                return None
            
            # Get recent samples
            batch_data = self.db_manager.sample_batch(
                batch_size=min(1000, total_samples),
                segment_length=1,
                agent_type="diagnostics",
                mode="latest"
            )
            
            if batch_data is None or len(batch_data['state']) == 0:
                return None
            
            # Extract states (each is a single timestep)
            states = np.array([seq[0] for seq in batch_data['state']])
            
            self.last_training_count = total_samples
            
            logging.info(f"Retrieved {len(states)} samples for threshold calculation")
            return states
            
        except Exception as e:
            logging.error(f"Error getting training data: {e}")
            return None
    
    def train_model(self, training_data: np.ndarray) -> Dict[str, Any]:
        """
        Calculate thresholds from data.
        
        Args:
            training_data: Array of sensor readings, shape (n_samples, n_channels)
            
        Returns:
            Dictionary of calculated thresholds per channel
        """
        try:
            if not self.use_dynamic:
                return {'status': 'static thresholds used'}
            
            # Get channel names from config
            channel_names = self.config.get('model_input', {}).get('channels', [])
            
            # Calculate percentile for each channel
            new_thresholds = {}
            for i, channel_name in enumerate(channel_names):
                if i < training_data.shape[1]:
                    threshold = np.percentile(training_data[:, i], self.percentile)
                    new_thresholds[channel_name] = float(threshold)
            
            # Update current thresholds
            self.current_thresholds.update(new_thresholds)
            
            metrics = {
                'thresholds': new_thresholds,
                'samples_used': len(training_data),
                'percentile': self.percentile
            }
            
            logging.info(f"Calculated thresholds: {new_thresholds}")
            
            return metrics
            
        except Exception as e:
            logging.error(f"Error calculating thresholds: {e}")
            return {'error': str(e)}
    
    def eval_model(self) -> Dict[str, Any]:
        """
        Evaluate threshold coverage.
        
        Returns:
            Statistics on how many samples exceed thresholds
        """
        try:
            # Get small sample for evaluation
            eval_data = self.get_training_data()
            if eval_data is None:
                return {'error': 'No evaluation data'}
            
            eval_data = eval_data[:min(100, len(eval_data))]
            
            channel_names = self.config.get('model_input', {}).get('channels', [])
            violations = {}
            
            for i, channel_name in enumerate(channel_names):
                if channel_name in self.current_thresholds and i < eval_data.shape[1]:
                    threshold = self.current_thresholds[channel_name]
                    exceeds = np.sum(eval_data[:, i] > threshold)
                    violations[channel_name] = {
                        'violation_count': int(exceeds),
                        'violation_rate': float(exceeds / len(eval_data)),
                        'threshold': threshold
                    }
            
            return {
                'violations': violations,
                'eval_samples': len(eval_data)
            }
            
        except Exception as e:
            logging.error(f"Error evaluating thresholds: {e}")
            return {'error': str(e)}
    
    def save_model(self, model_metrics: Dict[str, Any], eval_results: Dict[str, Any]):
        """
        Save thresholds to file.
        """
        try:
            os.makedirs('/app/models', exist_ok=True)
            
            threshold_file = '/app/models/current_thresholds.json'
            
            data = {
                'thresholds': self.current_thresholds,
                'metrics': model_metrics,
                'evaluation': eval_results,
                'timestamp': time.time()
            }
            
            with open(threshold_file, 'w') as f:
                json.dump(data, f, indent=2)
            
            logging.info(f"Saved thresholds to {threshold_file}")
            
        except Exception as e:
            logging.error(f"Error saving thresholds: {e}")

Key points:

Use get_size() to check data availability
Use sample_batch() to retrieve recent data
Track last_training_count to detect new data
Store results to filesystem for inference thread

Step 5: Implement Inference Thread

class ThresholdMLInferenceThread(MLInferenceThreadBase):
    """
    Checks sensor values against thresholds and publishes alerts.
    """
    
    def __init__(self, agent_id: str, config: Dict[str, Any]):
        self.thresholds = config.get('static_thresholds', {})
        self.use_dynamic = config.get('use_dynamic_thresholds', False)
        
        super().__init__(agent_id, config)
    
    def load_model(self):
        """Load current thresholds from file."""
        try:
            threshold_file = '/app/models/current_thresholds.json'
            
            if not os.path.exists(threshold_file):
                logging.warning("No threshold file found, using static thresholds")
                return
            
            with open(threshold_file, 'r') as f:
                data = json.load(f)
            
            if self.use_dynamic and 'thresholds' in data:
                self.thresholds = data['thresholds']
                logging.info(f"Loaded dynamic thresholds: {self.thresholds}")
        
        except json.JSONDecodeError as e:
            logging.error(f"Error parsing threshold file: {e}")
        except Exception as e:
            logging.error(f"Error loading thresholds: {e}")
    
    def parse_inference_request(self, message_data, topic, partition, offset) -> Optional[Dict[str, Any]]:
        """
        Parse sensor message for inference.
        
        Args:
            message_data: Filtered message dict
            topic: Kafka topic
            partition: Kafka partition
            offset: Kafka offset
            
        Returns:
            Parsed sensor data or None
        """
        try:
            channels = message_data.get('channels', {})
            
            if not channels:
                return None
            
            return {
                'values': channels,
                'timestamp': message_data.get('timestamp', time.time())
            }
            
        except Exception as e:
            logging.error(f"Error parsing inference request: {e}")
            return None
    
    def perform_inference(self, inference_request: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """
        Check values against thresholds.
        
        Args:
            inference_request: Parsed sensor data
            
        Returns:
            Alert information or None
        """
        try:
            values = inference_request['values']
            violations = {}
            has_violation = False
            
            for channel_name, value in values.items():
                if channel_name in self.thresholds:
                    threshold = self.thresholds[channel_name]
                    
                    if value > threshold:
                        violations[channel_name] = {
                            'value': float(value),
                            'threshold': float(threshold),
                            'exceeded_by': float(value - threshold)
                        }
                        has_violation = True
            
            if has_violation:
                logging.warning(f"Threshold violations detected: {violations}")
            
            return {
                'has_violation': has_violation,
                'violations': violations,
                'timestamp': inference_request['timestamp'],
                'all_values': values,
                'status': 'success'
            }
            
        except Exception as e:
            logging.error(f"Error performing inference: {e}")
            return None
    
    def _store_inference_result(self, inference_request: Dict[str, Any], inference_result: Dict[str, Any]):
        """
        Store inference result to database.
        
        For the threshold agent, we primarily publish alerts to Kafka rather than
        storing every inference result. However, this method can be extended to
        store violation history for analysis.
        
        Args:
            inference_request: The original inference request with sensor values
            inference_result: The inference result containing violation information
        """
        try:
            # For threshold agent, we don't need to store every inference result
            # since the base data is already stored by DataIngestThread.
            # We only log violations for monitoring purposes.
            if inference_result.get('has_violation', False):
                logging.debug(f"Threshold violation will be published to Kafka: "
                            f"{len(inference_result['violations'])} channels violated")
            
            # Optional: Store violation history to database for trend analysis
            # This could be implemented using db_manager.record_prediction() if needed
            
        except Exception as e:
            logging.error(f"Error in _store_inference_result: {e}")
    
    def process_message(self, message, topic, partition, offset) -> Tuple[bool, List[Tuple]]:
        """
        Process message and return alerts.
        
        Overrides base class to customize output format.
        """
        try:
            # Parse message
            if isinstance(message, bytes):
                message = message.decode('utf-8')
            
            message_data = json.loads(message)
            
            # Apply channel filtering
            if self.channel_filter:
                filtered_result = self.channel_filter.filter_channels(message_data)
                if filtered_result is None:
                    return True, []
                
                channel_names, channel_values = filtered_result
                message_data['channels'] = dict(zip(channel_names, channel_values))
            
            # Parse and perform inference
            inference_request = self.parse_inference_request(message_data, topic, partition, offset)
            
            if inference_request is None:
                return False, []
            
            inference_result = self.perform_inference(inference_request)
            
            if inference_result is None:
                return False, []
            
            # Store inference result (required abstract method)
            self._store_inference_result(inference_request, inference_result)
            
            # Only publish if there are violations
            if not inference_result['has_violation']:
                return True, []
            
            # Create alert message
            alert_channels = {
                'agent_id': self.agent_id,
                'has_violation': True,
                'violation_count': len(inference_result['violations'])
            }
            
            # Add violation details
            for channel_name, violation_info in inference_result['violations'].items():
                alert_channels[f'{channel_name}_value'] = violation_info['value']
                alert_channels[f'{channel_name}_threshold'] = violation_info['threshold']
                alert_channels[f'{channel_name}_exceeded_by'] = violation_info['exceeded_by']
            
            output_message = {
                'timestamp': time.time(),
                'channels': alert_channels
            }
            
            kafka_topic = self.producer.sanitize_topic_name(self.output_topic)
            return True, [(kafka_topic, json.dumps(output_message))]
            
        except Exception as e:
            logging.error(f"Error processing message: {e}")
            return False, []

Key points:

Load thresholds from file (saved by training thread)
Only publish alerts for violations (not every message)
Include violation details in output
Override process_message() for custom output logic
Implement _store_inference_result() as required abstract method

Step 6: Create Agent Class

class ThresholdAgent(AgentBase):
    """
    Agent for threshold-based anomaly detection.
    """
    
    def __init__(self, config_path: str = None, config_key: str = None):
        super().__init__("ThresholdAgent")
        
        # Load configuration
        if config_path:
            config_loader = ConfigLoader(config_path)
            if config_key is None:
                config_key = "threshold_agent1"
            self.agent_config = config_loader.config.get(config_key, {})
            
            self.enabled_threads = self.agent_config.get('enabled_threads', ['ingest', 'training', 'inference'])
        else:
            raise ValueError("config_path is required")
        
        # Add agent_id to config
        self.agent_config['agent_id'] = self.agent_id
        
        logging.info(f"ThresholdAgent initialized with config: {config_key}")
        logging.info(f"Enabled threads: {self.enabled_threads}")
    
    def create_data_ingest_component(self):
        """Create data ingestion thread."""
        if 'ingest' in self.enabled_threads:
            return ThresholdDataIngestThread(self.agent_id, self.agent_config)
        return None
    
    def create_ml_training_component(self):
        """Create ML training thread."""
        if 'training' in self.enabled_threads:
            return ThresholdMLTrainingThread(self.agent_id, self.agent_config)
        return None
    
    def create_ml_inference_component(self):
        """Create ML inference thread."""
        if 'inference' in self.enabled_threads:
            return ThresholdMLInferenceThread(self.agent_id, self.agent_config)
        return None


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser()
    parser.add_argument("--agent_config", help="Config key", type=str, default='threshold_agent1')
    args = parser.parse_args()
    
    setup_logging()
    
    config_path = os.getenv('CONFIG_PATH', '/app/config.yaml')
    
    try:
        agent = ThresholdAgent(config_path, args.agent_config)
        agent.start()
    except KeyboardInterrupt:
        logging.info("Shutting down threshold agent...")
    except Exception as e:
        logging.error(f"Error running threshold agent: {e}")
        raise


if __name__ == "__main__":
    main()

Step 7: Add Agent to Package

Update smocs/agents/__init__.py:

__all__ = ["AutoencoderAgent", "RLControlAgent", "ThresholdAgent"]

def __getattr__(name):
    if name == "AutoencoderAgent":
        from .autoencoder_agent import AutoencoderAgent
        return AutoencoderAgent
    elif name == "RLControlAgent":
        from .rl_control_agent import RLControlAgent
        return RLControlAgent
    elif name == "ThresholdAgent":
        from .threshold_agent import ThresholdAgent
        return ThresholdAgent
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

Step 8: Add Configuration

Add to orchestration/config.yaml:

threshold_agent1:
  # Threading
  enabled_threads: ['ingest', 'training', 'inference']
  
  # Data specification
  model_input:
    channels:
      - state_0
      - state_1
      - state_2
    bounds:
      - [-1.0, 1.0]
      - [-1.0, 1.0]
      - [-8.0, 8.0]
  
  # Thresholds
  static_thresholds:
    state_0: 0.8
    state_1: 0.8
    state_2: 6.0
  
  use_dynamic_thresholds: true
  threshold_percentile: 95.0
  min_training_samples: 500
  
  # Topics
  kafka_topics:
    input: "gymnasium-output"
    output: "threshold-alerts"
    training_output: "threshold-updates"

Step 9: Create Requirements File

Create orchestration/containers/agent/threshold-requirements.txt:

numpy==1.24.3

Important: This file tells Docker which additional Python packages to install for the threshold agent. The base requirements (Kafka, MySQL, YAML) are already installed via base-requirements.txt, but we need NumPy for array operations.

Step 10: Add Docker Compose Service

Add to orchestration/docker-compose.yml:

threshold-agent1:
  extends: agent
  profiles: ["threshold1"]
  environment:
    AGENT_TYPE: threshold
    AGENT_CONFIG: threshold_agent1
    SMOCS_LOG_LEVEL: INFO
  ports:
    - "3309:3306"
  volumes:
    - mysql-data-threshold1:/var/lib/mysql

Add volume to volumes section:

volumes:
  # ... existing volumes ...
  mysql-data-threshold1:

Step 11: Test the Agent

1. Update .env:

COMPOSE_PROFILES=gymnasium,threshold1

2. Build and start:

docker compose build threshold-agent1
docker compose up threshold-agent1 gymnasium-kafka-controller

3. Verify logs:

# Check agent startup
docker compose logs threshold-agent1 | head -30

# Watch for data ingestion
docker compose logs -f threshold-agent1 | grep "Stored sensor data"

# Watch for threshold violations
docker compose logs -f threshold-agent1 | grep "violations detected"

4. Check database:

docker exec -it threshold-agent1 mysql -u root -p

USE agentdb;
SELECT COUNT(*) FROM agent_inferences;
SELECT * FROM agent_inferences ORDER BY id DESC LIMIT 5;

5. Monitor training progress:

Wait for approximately 500 samples to accumulate (this may take a few minutes depending on your data rate), then check for training logs:

docker compose logs threshold-agent1 | grep "Training"

You should see messages like:

Database contains 500 samples (need 500)
Retrieved 500 samples for threshold calculation
Calculated thresholds: {'state_0': 0.82, 'state_1': 0.79, 'state_2': 5.95}

Complete Agent Summary

What the agent does:

Ingests sensor data from Kafka
Stores to MySQL database
Calculates dynamic thresholds from historical data
Checks realtime values against thresholds
Publishes alerts when violations occur

Files created:

smocs/agents/threshold_agent.py (~350 lines)
Configuration in config.yaml
threshold-requirements.txt
Docker compose service

Demonstrated:

Three thread architecture
Database integration with DBManager
Kafka consumption and production
Configuration driven behavior
Channel filtering
Error handling
Abstract method implementation

Next Steps

Add preprocessing: Use preprocessing pipeline for normalization
Add visualization: Create InfluxDB dashboard for alerts
Add hysteresis: Prevent alert flapping
Store violation history: Extend _store_inference_result() to save violations to database

Overview​

Prerequisites​

Step 1: Plan Your Agent​

Agent Responsibilities​

Configuration Design​

Step 2: Create Agent File​

Step 3: Implement Data Ingest Thread​

Step 4: Implement Training Thread​

Step 5: Implement Inference Thread​

Step 6: Create Agent Class​

Step 7: Add Agent to Package​

Step 8: Add Configuration​

Step 9: Create Requirements File​

Step 10: Add Docker Compose Service​

Step 11: Test the Agent​

Complete Agent Summary​

Next Steps​

Overview

Prerequisites

Step 1: Plan Your Agent

Agent Responsibilities

Configuration Design

Step 2: Create Agent File

Step 3: Implement Data Ingest Thread

Step 4: Implement Training Thread

Step 5: Implement Inference Thread

Step 6: Create Agent Class

Step 7: Add Agent to Package

Step 8: Add Configuration

Step 9: Create Requirements File

Step 10: Add Docker Compose Service

Step 11: Test the Agent

Complete Agent Summary

Next Steps